package indi.zhaosheng.rocktmq.producer;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author muluo
 * @description
 * @date 2020/5/5 15:46
 */
public abstract class BaseNormalProducer extends BaseProducer {

    private static final Logger logger = LoggerFactory.getLogger(BaseNormalProducer.class);

    protected BaseNormalProducer(String topic, String producerGroup) {
        super(topic, producerGroup);
    }

    public boolean sendMsg(String tag, Object msg) {
        Message message = new Message(topic, tag, JSON.toJSONBytes(msg));
        try {
            SendResult sendResult = defaultMQProducer.send(message);
            logger.info("msg send to topic:{}, msgId:{}, status:{}",
                    this.topic, sendResult.getMsgId(), sendResult.getSendStatus());
            return true;
        } catch (Exception e) {
            logger.error("msg send to topic:{}, body:{}, error context:{}",
                    this.topic, JSON.toJSONString(message), e);
        }
        return false;
    }

    public boolean sendMsgAndRetryWhenSendField(String tag, Object msg, int retryTimes, long timeout) {
        Message message = new Message(topic, tag, JSON.toJSONBytes(msg));
        defaultMQProducer.setRetryTimesWhenSendFailed(retryTimes);
        try {
            SendResult sendResult = defaultMQProducer.send(message, timeout);
            logger.info("msg send to topic:{}, msgId:{}, status:{}",
                    this.topic, sendResult.getMsgId(), sendResult.getSendStatus());
            return true;
        } catch (Exception e) {
            logger.error("msg send to topic:{}, body:{}, error context:{}",
                    this.topic, JSON.toJSONString(message), e);
        }
        return false;
    }


}
